// This file is a part of ChaosDB.
// Copyright (C) 2012 ChaoJianPeng
// email:chaojianpeng@gmail.com
//
// ChaosDB is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// any later version.
//
// ChaosDB is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

/*
 * bio.cpp
 *
 *
 *  Created on: 2012-3-5
 *      Author: chao
 */

#include "bio.h"
#include "conf.h"
#include "bdb_api.h"
#include "string.h"
#include "logger.h"
#include "cmalloc.h"
#include "respond.h"
#include <stdio.h>
#include <pthread.h>
#include <zmq.h>
#include <stdlib.h>

pthread_t ntid;
void *ctx, *pair;

char* cstrcat(char* s1, char* s2, int total_len) {
	char* res = (char*) malloc(total_len + 1);
	int j = -1;
	for (int i = 0; s1[i] != '\0'; ++i) {
		res[++j] = s1[i];
	}
	for (int i = 0; s2[i] != '\0'; ++i) {
		res[++j] = s2[i];
	}
	res[++j] = '\0';
	return res;
}

void cfwrite(void* arg) {

	fw_args* _arg = (fw_args*) arg;
	char* filepath = cstrcat("data/", _arg->key, _arg->keyLen + 5);
	// 写入
	if (_arg->writeOrdelete == BIO_WRITE) {
		FILE* file = fopen(filepath, "wb");
		if (file == NULL) {
			char errorStr[100];
			sprintf(errorStr, "filepath Error:%s", _arg->key);
			systemLogger(errorStr);
			return;
		}
		// 错误需要标记
		size_t size = _arg->valueLen;
		fwrite(&size, PREFIX_SIZE, 1, file);
		fwrite(_arg->value, size, 1, file);

		fclose(file);

	}
	// 删除
	else {
		remove(filepath);
	}
	// 清空一下内存,所有进来这里的都是复制来的,都要清理
	free(filepath);
	free(_arg->key);
	if (_arg->value != NULL) {
		free(_arg->value);
	}
}

void *write_thread(void *ctx) {
	void *pair;
	zmq_msg_t msg;

	pair = zmq_socket(ctx, ZMQ_PAIR);
	zmq_connect(pair, "inproc://workers");

	while (1) {
		zmq_msg_init(&msg);
		zmq_recv(pair, &msg, 0);
		cfwrite(zmq_msg_data(&msg));
		zmq_msg_close(&msg);
	}

	zmq_close(pair);

	return (void *) 0;
}

/**
 * 开启一个线程
 */
void bioInit() {
	if (STORAGE_TYPE == DEFAULT_BIO) {
		ctx = zmq_init(0);
		pair = zmq_socket(ctx, ZMQ_PAIR);
		zmq_bind(pair, "inproc://workers");
		pthread_create(&ntid, NULL, write_thread, ctx);
	} else if (STORAGE_TYPE == BERKELET_DB_BIO) {
		bdbInit();
	}
}

void bioJoin() {
	pthread_join(ntid, NULL);
}

/**
 * 异步写文件
 */
void fileWrite(char* key, void* value, int keyLen, size_t valueLen) {

	//  复制对象
	char* k = (char*) malloc(keyLen + 1);
	void* v = malloc(valueLen);
	memcpy(k, key, keyLen + 1);
	memcpy(v, value, valueLen);

	// 一个key一个文件
	if (STORAGE_TYPE == DEFAULT_BIO) {
		fw_args arg;
		arg.key = k;
		arg.value = v;
		arg.keyLen = keyLen;
		arg.valueLen = valueLen;
		arg.writeOrdelete = BIO_WRITE;

		// 发送消息
		zmq_msg_t msg;
		zmq_msg_init_size(&msg, sizeof(fw_args));
		memcpy(zmq_msg_data(&msg), &arg, sizeof(fw_args));
		zmq_send(pair, &msg, 0);
		zmq_msg_close(&msg);
	}
	// berkeleyDB写入
	else if (STORAGE_TYPE == BERKELET_DB_BIO) {
		bdbWrite(k, keyLen, v, valueLen);
	}
}

void* fileRead(char* key, int len) {
	void* buff;
	if (STORAGE_TYPE == DEFAULT_BIO) {
		char* filepath = cstrcat("data/", key, len + 5);
		FILE* file = fopen(filepath, "rb");
		if (file == NULL)
			return NULL;
		size_t size;
		fread(&size, PREFIX_SIZE, 1, file);

		buff = cmalloc(size);
		fread(buff, size, 1, file);
		fclose(file);
	} else if (STORAGE_TYPE == BERKELET_DB_BIO) {
		buff = bdbRead(key, len);
	}
	return buff;
}

void fileDel(char* key, int len) {
	//  复制对象
	char* k = (char*) malloc(len + 1);
	memcpy(k, key, len + 1);

	if (STORAGE_TYPE == DEFAULT_BIO) {
		fw_args arg;
		arg.key = k;
		arg.keyLen = len;
		arg.writeOrdelete = BIO_DELETE;
		arg.valueLen = 0;
		arg.value = NULL;

		// 发送消息
		zmq_msg_t msg;
		zmq_msg_init_size(&msg, sizeof(fw_args));
		memcpy(zmq_msg_data(&msg), &arg, sizeof(fw_args));
		zmq_send(pair, &msg, 0);
		zmq_msg_close(&msg);
	} else if (STORAGE_TYPE == BERKELET_DB_BIO) {
		bdbDelete(key, len);
	}
}
